/**
 * Copyright 1999-2015 dangdang.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * </p>
 */

package spring.cloud.tasks.api.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import spring.cloud.tasks.api.database.entity.DashboardHistoryDto;
import spring.cloud.tasks.api.database.entity.StatisticsDto;
import spring.cloud.tasks.api.database.repository.DashboardHistoryDtoRepository;
import spring.cloud.tasks.api.database.service.StatisticsDtoService;
import spring.cloud.tasks.api.domain.*;
import spring.cloud.tasks.api.service.DashboardService;
import spring.cloud.tasks.api.service.StatisticsRefreshService;
import spring.cloud.tasks.api.service.TaskService;
import spring.cloud.tasks.api.service.ZooKeeperService;
import spring.cloud.tasks.api.service.helper.DashboardServiceHelper;
import spring.cloud.tasks.api.utils.ExecutorNodePath;
import spring.cloud.tasks.api.utils.ResetCountType;
import spring.cloud.tasks.api.utils.StatisticsTableKeyConstant;
import spring.cloud.tasks.api.utils.TaskNodePath;
import spring.cloud.tasks.api.zookeeper.ZooKeeper;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

public class DashboardServiceImpl implements DashboardService {

    private static final Logger log = LoggerFactory.getLogger(DashboardServiceImpl.class);

    @Autowired
    private StatisticsDtoService statisticsDtoService;

    @Autowired
    private ZooKeeperService zooKeeperService;

    @Autowired
    private TaskService taskService;

    @Autowired
    private StatisticsRefreshService statisticsRefreshService;

    private ExecutorService updateStatisticsThreadPool;

    @Autowired
    private DashboardHistoryDtoRepository dashboardHistoryDtoRepository;

    @PostConstruct
    public void init() {
        initUpdateStatisticsThreadPool();
    }

    @PreDestroy
    public void destroy() {
        if (updateStatisticsThreadPool != null) {
            updateStatisticsThreadPool.shutdownNow();
        }
    }

    private void initUpdateStatisticsThreadPool() {
        updateStatisticsThreadPool = Executors.newSingleThreadExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                String name = "single-update-statistics";
                Thread t = new Thread(r, name);
                if (t.isDaemon()) {
                    t.setDaemon(false);
                }
                if (t.getPriority() != Thread.NORM_PRIORITY) {
                    t.setPriority(Thread.NORM_PRIORITY);
                }
                return t;
            }
        });
    }

    @Override
    public int executorInDockerCount(String zkList) throws RuntimeException {
        return getCountFromDB(StatisticsTableKeyConstant.EXECUTOR_IN_DOCKER_COUNT, zkList);
    }

    @Override
    public int executorNotInDockerCount(String zkList) throws RuntimeException {
        return getCountFromDB(StatisticsTableKeyConstant.EXECUTOR_NOT_IN_DOCKER_COUNT, zkList);
    }

    @Override
    public int jobCount(String zkList) throws RuntimeException {
        return getCountFromDB(StatisticsTableKeyConstant.JOB_COUNT, zkList);
    }

    private int getCountFromDB(String name, String zkList) {
        StatisticsDto ss = statisticsDtoService.findStatisticsByNameAndZkList(name, zkList);
        if (ss == null || StringUtils.isBlank(ss.getResult())) {
            return 0;
        }

        String result = ss.getResult();
        try {
            Integer count = JSON.parseObject(result, new TypeReference<Integer>() {
            });
            return count == null ? 0 : count;
        } catch (Exception e) {
            log.error("exception throws during get count from DB. name:" + name, e);
            return 0;
        }
    }

    @Override
    public StatisticsDto top10FailureJob(String zklist) throws RuntimeException {
        return statisticsDtoService
                .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_FAIL_JOB, zklist);
    }

    @Override
    public String top10FailureJobByAllZkCluster() throws RuntimeException {
        List<JobStatistics> jobStatisticsList = new ArrayList<>();
        Collection<ZooKeeperCluster> zkClusterList = zooKeeperService.getOnlineZooKeeperClusterList();
        for (ZooKeeperCluster zkCluster : zkClusterList) {
            StatisticsDto saturnStatistics = top10FailureJob(zkCluster.getZooKeeperAddress());
            if (saturnStatistics != null) {
                String result = saturnStatistics.getResult();
                List<JobStatistics> tempList = JSON.parseArray(result, JobStatistics.class);
                if (tempList != null) {
                    jobStatisticsList.addAll(tempList);
                }
            }
        }
        jobStatisticsList = DashboardServiceHelper.sortJobByAllTimeFailureRate(jobStatisticsList);
        List<JobStatistics> top10FailJob = jobStatisticsList
                .subList(0, jobStatisticsList.size() > 9 ? 10 : jobStatisticsList.size());
        return JSON.toJSONString(top10FailJob);
    }

    @Override
    public StatisticsDto top10FailureExecutor(String zklist) throws RuntimeException {
        return statisticsDtoService
                .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_FAIL_EXECUTOR, zklist);
    }

    @Override
    public String top10FailureExecutorByAllZkCluster() throws RuntimeException {
        List<ExecutorStatistics> executorStatisticsList = new ArrayList<>();
        Collection<ZooKeeperCluster> zkClusterList = zooKeeperService.getOnlineZooKeeperClusterList();
        for (ZooKeeperCluster zkCluster : zkClusterList) {
            StatisticsDto saturnStatistics = top10FailureExecutor(zkCluster.getZooKeeperAddress());
            if (saturnStatistics != null) {
                String result = saturnStatistics.getResult();
                List<ExecutorStatistics> tempList = JSON.parseArray(result, ExecutorStatistics.class);
                if (tempList != null) {
                    executorStatisticsList.addAll(tempList);
                }
            }
        }
        executorStatisticsList = DashboardServiceHelper.sortExecutorByFailureRate(executorStatisticsList);
        List<ExecutorStatistics> top10FailureExecutor = executorStatisticsList
                .subList(0, executorStatisticsList.size() > 9 ? 10 : executorStatisticsList.size());
        return JSON.toJSONString(top10FailureExecutor);
    }

    @Override
    public StatisticsDto top10AactiveJob(String zklist) throws RuntimeException {
        return statisticsDtoService
                .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_ACTIVE_JOB, zklist);
    }

    @Override
    public String top10AactiveJobByAllZkCluster() throws RuntimeException {
        List<JobStatistics> jobStatisticsList = new ArrayList<>();
        Collection<ZooKeeperCluster> zkClusterList = zooKeeperService.getOnlineZooKeeperClusterList();
        for (ZooKeeperCluster zkCluster : zkClusterList) {
            StatisticsDto saturnStatistics = top10AactiveJob(zkCluster.getZooKeeperAddress());
            if (saturnStatistics != null) {
                String result = saturnStatistics.getResult();
                List<JobStatistics> tempList = JSON.parseArray(result, JobStatistics.class);
                if (tempList != null) {
                    jobStatisticsList.addAll(tempList);
                }
            }
        }
        jobStatisticsList = DashboardServiceHelper.sortJobByDayProcessCount(jobStatisticsList);
        List<JobStatistics> top10AactiveJob = jobStatisticsList
                .subList(0, jobStatisticsList.size() > 9 ? 10 : jobStatisticsList.size());
        return JSON.toJSONString(top10AactiveJob);
    }

    @Override
    public StatisticsDto top10LoadExecutor(String zklist) throws RuntimeException {
        return statisticsDtoService
                .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_LOAD_EXECUTOR, zklist);
    }

    @Override
    public String top10LoadExecutorByAllZkCluster() throws RuntimeException {
        List<ExecutorStatistics> executorStatisticsList = new ArrayList<>();
        Collection<ZooKeeperCluster> zkClusterList = zooKeeperService.getOnlineZooKeeperClusterList();
        for (ZooKeeperCluster zkCluster : zkClusterList) {
            StatisticsDto saturnStatistics = top10LoadExecutor(zkCluster.getZooKeeperAddress());
            if (saturnStatistics != null) {
                String result = saturnStatistics.getResult();
                List<ExecutorStatistics> tempList = JSON.parseArray(result, ExecutorStatistics.class);
                if (tempList != null) {
                    executorStatisticsList.addAll(tempList);
                }
            }
        }
        executorStatisticsList = DashboardServiceHelper.sortExecutorByLoadLevel(executorStatisticsList);
        List<ExecutorStatistics> top10LoadExecutor = executorStatisticsList
                .subList(0, executorStatisticsList.size() > 9 ? 10 : executorStatisticsList.size());
        return JSON.toJSONString(top10LoadExecutor);
    }

    @Override
    public StatisticsDto top10LoadJob(String zklist) throws RuntimeException {
        return statisticsDtoService
                .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_LOAD_JOB, zklist);
    }

    @Override
    public String top10LoadJobByAllZkCluster() throws RuntimeException {
        List<JobStatistics> jobStatisticsList = new ArrayList<>();
        Collection<ZooKeeperCluster> zkClusterList = zooKeeperService.getOnlineZooKeeperClusterList();
        for (ZooKeeperCluster zkCluster : zkClusterList) {
            StatisticsDto saturnStatistics = top10LoadJob(zkCluster.getZooKeeperAddress());
            if (saturnStatistics != null) {
                String result = saturnStatistics.getResult();
                List<JobStatistics> tempList = JSON.parseArray(result, JobStatistics.class);
                if (tempList != null) {
                    jobStatisticsList.addAll(tempList);
                }
            }
        }
        jobStatisticsList = DashboardServiceHelper.sortJobByLoadLevel(jobStatisticsList);
        List<JobStatistics> top10LoadJob = jobStatisticsList
                .subList(0, jobStatisticsList.size() > 9 ? 10 : jobStatisticsList.size());
        return JSON.toJSONString(top10LoadJob);
    }

    @Override
    public StatisticsDto top10UnstableDomain(String zklist) throws RuntimeException {
        return statisticsDtoService
                .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_UNSTABLE_DOMAIN, zklist);
    }

    @Override
    public String top10UnstableDomainByAllZkCluster() throws RuntimeException {
        List<DomainStatistics> domainStatisticsList = new ArrayList<>();
        Collection<ZooKeeperCluster> zkClusterList = zooKeeperService.getOnlineZooKeeperClusterList();
        for (ZooKeeperCluster zkCluster : zkClusterList) {
            StatisticsDto saturnStatistics = top10UnstableDomain(zkCluster.getZooKeeperAddress());
            if (saturnStatistics != null) {
                String result = saturnStatistics.getResult();
                List<DomainStatistics> tempList = JSON.parseArray(result, DomainStatistics.class);
                if (tempList != null) {
                    domainStatisticsList.addAll(tempList);
                }
            }
        }
        domainStatisticsList = DashboardServiceHelper.sortDomainByShardingCount(domainStatisticsList);
        List<DomainStatistics> top10UnstableDomain = domainStatisticsList
                .subList(0, domainStatisticsList.size() > 9 ? 10 : domainStatisticsList.size());
        return JSON.toJSONString(top10UnstableDomain);
    }

    @Override
    public StatisticsDto allProcessAndErrorCountOfTheDay(String zklist) throws RuntimeException {
        return statisticsDtoService
                .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.DOMAIN_PROCESS_COUNT_OF_THE_DAY, zklist);
    }

    @Override
    public String allProcessAndErrorCountOfTheDayByAllZkCluster() throws RuntimeException {
        long count = 0;
        long error = 0;
        Collection<ZooKeeperCluster> zkClusterList = zooKeeperService.getOnlineZooKeeperClusterList();
        for (ZooKeeperCluster zkCluster : zkClusterList) {
            StatisticsDto saturnStatistics = allProcessAndErrorCountOfTheDay(zkCluster.getZooKeeperAddress());
            if (saturnStatistics != null) {
                String result = saturnStatistics.getResult();
                ZkStatistics temp = JSON.parseObject(result, ZkStatistics.class);
                if (temp != null) {
                    count += temp.getCount();
                    error += temp.getError();
                }
            }
        }
        return JSON.toJSONString(new ZkStatistics(count, error));
    }

    @Override
    public StatisticsDto top10FailureDomain(String zklist) throws RuntimeException {
        return statisticsDtoService
                .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_FAIL_DOMAIN, zklist);
    }

    @Override
    public String top10FailureDomainByAllZkCluster() throws RuntimeException {
        List<DomainStatistics> domainStatisticsList = new ArrayList<>();
        Collection<ZooKeeperCluster> zkClusterList = zooKeeperService.getOnlineZooKeeperClusterList();
        for (ZooKeeperCluster zkCluster : zkClusterList) {
            StatisticsDto saturnStatistics = top10FailureDomain(zkCluster.getZooKeeperAddress());
            if (saturnStatistics != null) {
                String result = saturnStatistics.getResult();
                List<DomainStatistics> tempList = JSON.parseArray(result, DomainStatistics.class);
                if (tempList != null) {
                    domainStatisticsList.addAll(tempList);
                }
            }
        }
        domainStatisticsList = DashboardServiceHelper.sortDomainByAllTimeFailureRate(domainStatisticsList);
        List<DomainStatistics> top10FailureDomain = domainStatisticsList
                .subList(0, domainStatisticsList.size() > 9 ? 10 : domainStatisticsList.size());
        return JSON.toJSONString(top10FailureDomain);
    }

    @Override
    public void cleanShardingCount(String namespace) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(namespace);
        zooKeeper.update(ExecutorNodePath.SHARDING_COUNT_PATH, "0");
        asyncForceRefreshStatistics(namespace);
    }

    @Override
    public void cleanOneJobAnalyse(String namespace, String jobName) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(namespace);
        updateResetValue(zooKeeper, jobName, ResetCountType.RESET_ANALYSE);
        resetOneJobAnalyse(jobName, zooKeeper);
        asyncForceRefreshStatistics(namespace);
    }

    @Override
    public void cleanAllJobAnalyse(String namespace) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(namespace);
        List<String> jobs = taskService.getUnSystemJobNames(namespace);
        for (String job : jobs) {
            resetOneJobAnalyse(job, zooKeeper);
            updateResetValue(zooKeeper, job, ResetCountType.RESET_ANALYSE);
        }
        asyncForceRefreshStatistics(namespace);
    }

    @Override
    public void cleanAllJobExecutorCount(String namespace) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(namespace);
        List<String> jobs = taskService.getUnSystemJobNames(namespace);
        for (String job : jobs) {
            resetOneJobExecutorCount(job, zooKeeper);
            updateResetValue(zooKeeper, job, ResetCountType.RESET_SERVERS);
        }
        asyncForceRefreshStatistics(namespace);
    }

    @Override
    public void cleanOneJobExecutorCount(String namespace, String jobName) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(namespace);
        updateResetValue(zooKeeper, jobName, ResetCountType.RESET_SERVERS);
        resetOneJobExecutorCount(jobName, zooKeeper);
        asyncForceRefreshStatistics(namespace);
    }

    private void resetOneJobExecutorCount(String jobName, ZooKeeper zooKeeper) {
        if (zooKeeper.checkExists(TaskNodePath.getExecutorsNodePath(jobName))) {
            List<String> servers = zooKeeper.getChildren(TaskNodePath.getExecutorsNodePath(jobName));
            for (String server : servers) {
                zooKeeper.update(TaskNodePath.getProcessSucessCount(jobName, server), "0");
                zooKeeper.update(TaskNodePath.getProcessFailureCount(jobName, server), "0");
            }
        }
    }

    private void resetOneJobAnalyse(String jobName, ZooKeeper zooKeeper) {
        zooKeeper.update(TaskNodePath.getProcessCountPath(jobName), "0");
        zooKeeper.update(TaskNodePath.getErrorCountPath(jobName), "0");
    }

    private void updateResetValue(ZooKeeper zooKeeper, String job, String value) {
        String path = TaskNodePath.getAnalyseResetPath(job);
        zooKeeper.update(path, value.getBytes());
    }

    private void asyncForceRefreshStatistics(final String namespace) {
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                try {
                    NameSpaceConfiguration conf = zooKeeperService.getNameSpaceConfigurationByNameSpace(namespace);
                    if (conf != null) {
                        statisticsRefreshService.refresh(conf.getZooKeeperClusterKey(), false);
                    }
                } catch (Throwable t) {
                    log.error("async refresh statistics error", t);
                }
            }
        };
        updateStatisticsThreadPool.submit(runnable);
    }

    @Override
    public Map<String, Integer> loadDomainRankDistribution(String zkClusterKey) throws RuntimeException {
        Map<String, Integer> domainMap = new HashMap<>();
        if (zkClusterKey != null) {
            ZooKeeperCluster zkCluster = zooKeeperService.getZooKeeperCluster(zkClusterKey);
            if (zkCluster != null && !zkCluster.isOffline()) {
                for (NameSpaceConfiguration config : zkCluster.getNameSpaceConfigurationList()) {
                    Integer count = domainMap.get(config.getDegree());
                    if (null != config.getDegree()) {
                        domainMap.put(config.getDegree(), count == null ? 1 : count + 1);
                    }
                }
            }
        }
        return domainMap;
    }

    @Override
    public Map<String, Integer> loadDomainRankDistributionByAllZkCluster() throws RuntimeException {
        Map<String, Integer> domainMap = new HashMap<>();
        Collection<ZooKeeperCluster> zkClusterList = zooKeeperService.getOnlineZooKeeperClusterList();
        for (ZooKeeperCluster zkCluster : zkClusterList) {
            List<NameSpaceConfiguration> regCenterConfList = zkCluster.getNameSpaceConfigurationList();
            if (regCenterConfList != null) {
                for (NameSpaceConfiguration config : regCenterConfList) {
                    String degree = config.getDegree();
                    if (degree != null) {
                        Integer count = domainMap.get(degree);
                        domainMap.put(degree, count == null ? 1 : count + 1);
                    }
                }
            }
        }
        return domainMap;
    }

    @Override
    public Map<Integer, Integer> loadJobRankDistribution(String zklist) throws RuntimeException {
        StatisticsDto ss = statisticsDtoService
                .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.JOB_RANK_DISTRIBUTION, zklist);
        if (ss != null) {
            String result = ss.getResult();
            return JSON.parseObject(result, new TypeReference<Map<Integer, Integer>>() {
            });
        } else {
            return new HashMap<>();
        }
    }

    @Override
    public Map<Integer, Integer> loadJobRankDistributionByAllZkCluster() throws RuntimeException {
        Map<Integer, Integer> jobDegreeCountMap = new HashMap<>();
        Collection<ZooKeeperCluster> zkClusterList = zooKeeperService.getOnlineZooKeeperClusterList();
        for (ZooKeeperCluster zkCluster : zkClusterList) {
            String zkAddr = zkCluster.getZooKeeperAddress();
            Map<Integer, Integer> temp = loadJobRankDistribution(zkAddr);
            if (temp == null) {
                continue;
            }
            Iterator<Entry<Integer, Integer>> iterator = temp.entrySet().iterator();
            while (iterator.hasNext()) {
                Entry<Integer, Integer> next = iterator.next();
                Integer jobDegree = next.getKey();
                Integer count = next.getValue();
                if (jobDegree != null && count != null) {
                    if (jobDegreeCountMap.containsKey(jobDegree)) {
                        jobDegreeCountMap.put(jobDegree, jobDegreeCountMap.get(jobDegree) + count);
                    } else {
                        jobDegreeCountMap.put(jobDegree, count);
                    }
                }
            }
        }
        return jobDegreeCountMap;
    }

    @Override
    public Map<String, Long> versionDomainNumber(String currentZkAddr) throws RuntimeException {
        StatisticsDto ss = statisticsDtoService
                .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.VERSION_DOMAIN_NUMBER, currentZkAddr);
        if (ss != null) {
            String result = ss.getResult();
            return JSON.parseObject(result, new TypeReference<Map<String, Long>>() {
            });
        } else {
            return new HashMap<>();
        }
    }

    @Override
    public Map<String, Long> versionDomainNumberByAllZkCluster() throws RuntimeException {
        Map<String, Long> versionDomainNumberMap = new HashMap<>();
        Collection<ZooKeeperCluster> zkClusterList = zooKeeperService.getOnlineZooKeeperClusterList();
        for (ZooKeeperCluster zkCluster : zkClusterList) {
            String zkAddr = zkCluster.getZooKeeperAddress();
            Map<String, Long> temp = versionDomainNumber(zkAddr);
            if (temp == null) {
                continue;
            }
            Iterator<Entry<String, Long>> iterator = temp.entrySet().iterator();
            while (iterator.hasNext()) {
                Entry<String, Long> next = iterator.next();
                String version = next.getKey();
                Long domainNumber = next.getValue();
                if (version != null && domainNumber != null) {
                    if (versionDomainNumberMap.containsKey(version)) {
                        versionDomainNumberMap.put(version, versionDomainNumberMap.get(version) + domainNumber);
                    } else {
                        versionDomainNumberMap.put(version, domainNumber);
                    }
                }
            }
        }
        return versionDomainNumberMap;
    }

    @Override
    public Map<String, Long> versionExecutorNumber(String currentZkAddr) throws RuntimeException {
        StatisticsDto ss = statisticsDtoService
                .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.VERSION_EXECUTOR_NUMBER, currentZkAddr);
        if (ss != null) {
            String result = ss.getResult();
            return JSON.parseObject(result, new TypeReference<Map<String, Long>>() {
            });
        } else {
            return new HashMap<>();
        }
    }

    @Override
    public Map<String, Long> versionExecutorNumberByAllZkCluster() throws RuntimeException {
        Map<String, Long> versionExecutorNumberMap = new HashMap<>();
        Collection<ZooKeeperCluster> zkClusterList = zooKeeperService.getOnlineZooKeeperClusterList();
        for (ZooKeeperCluster zkCluster : zkClusterList) {
            String zkAddr = zkCluster.getZooKeeperAddress();
            Map<String, Long> temp = versionExecutorNumber(zkAddr);
            if (temp == null) {
                continue;
            }
            Iterator<Entry<String, Long>> iterator = temp.entrySet().iterator();
            while (iterator.hasNext()) {
                Entry<String, Long> next = iterator.next();
                String version = next.getKey();
                Long executorNumber = next.getValue();
                if (version != null && executorNumber != null) {
                    if (versionExecutorNumberMap.containsKey(version)) {
                        versionExecutorNumberMap.put(version, versionExecutorNumberMap.get(version) + executorNumber);
                    } else {
                        versionExecutorNumberMap.put(version, executorNumber);
                    }
                }
            }
        }
        return versionExecutorNumberMap;
    }

    @Override
    public void saveDashboardHistory(String zkCluster, String type, String topic, String content) {
        dashboardHistoryDtoRepository.createOrUpdateHistory(zkCluster, type, topic, content, new Date());
    }

    @Override
    public void batchSaveDashboardHistory(List<DashboardHistoryDto> dashboardHistories) {
        dashboardHistoryDtoRepository.batchCreateOrUpdateHistory(dashboardHistories);
    }

    @Override
    public List<DashboardHistoryDto> getDashboardDomainHistory(String zkCluster, String type, String topic, Date fromDate,
                                                               Date toDate) throws RuntimeException {
        List<String> zkClusters = new ArrayList<>();
        zkClusters.add(zkCluster);
        return dashboardHistoryDtoRepository
                .selectByZkClustersAndTypeAndTopicAndFromStartDateToEndDate(zkClusters, type, topic, fromDate, toDate);
    }

    @Override
    public Map<String, List> getDomainOperationHistory(String type, String topic, Date fromDate, Date toDate) {
        SimpleDateFormat sdf = new SimpleDateFormat("MM-dd");

        List<String> zkClusters = getAvailableZkClusterKey();

        if (zkClusters.isEmpty()) {
            Map<String, List> result = new HashMap<>(2);
            result.put("xAxis", null);
            result.put("yAxis", null);
            return result;
        }

        List<DashboardHistoryDto> dashboardHistories = dashboardHistoryDtoRepository
                .selectByZkClustersAndTypeAndTopicAndFromStartDateToEndDate(zkClusters, type, topic, fromDate, toDate);

        Calendar tmpCalendar = Calendar.getInstance();
        tmpCalendar.setTime(fromDate);
        Date tmpDate = tmpCalendar.getTime();

        List<String> resultDate = new ArrayList<>();
        List<Long> resultCount = new ArrayList<>();
        while (tmpDate.getTime() <= toDate.getTime()) {
            List<DashboardHistoryDto> tmpHistories = new ArrayList<>();
            for (DashboardHistoryDto history : dashboardHistories) {

                String d1 = sdf.format(history.getRecordDate().getTime());
                String d2 = sdf.format(tmpDate.getTime());

                if (StringUtils.equals(d1, d2)) {
                    tmpHistories.add(history);
                }
            }
            long count = 0;
            for (DashboardHistoryDto sumHistory : tmpHistories) {
                Map content = JSON.parseObject(sumHistory.getContent(), Map.class);
                Long tmpCount = Long.valueOf(String.valueOf(content.get("count")));
                count += tmpCount;
            }
            resultDate.add(sdf.format(tmpDate));
            resultCount.add(count);

            tmpCalendar.add(Calendar.DATE, 1);
            tmpDate = tmpCalendar.getTime();
        }

        Map<String, List> result = new HashMap<>(2);
        result.put("xAxis", resultDate);
        result.put("yAxis", resultCount);

        return result;


    }

    @Override
    public Map<String, List> getDomainCountHistory(String zkCluster, String type, String topic, Date fromDate,
                                                   Date toDate) {
        SimpleDateFormat sdf = new SimpleDateFormat("MM-dd");

        List<String> zkClusters = new ArrayList<>();

        if (StringUtils.isEmpty(zkCluster)) {
            zkClusters.addAll(getAvailableZkClusterKey());
        } else {
            zkClusters.add(zkCluster);
        }

        if (zkClusters.isEmpty()) {
            Map<String, List> result = new HashMap<>(2);
            result.put("xAxis", null);
            result.put("yAxis", null);
            return result;
        }

        List<DashboardHistoryDto> dashboardHistories = dashboardHistoryDtoRepository
                .selectByZkClustersAndTypeAndTopicAndFromStartDateToEndDate(zkClusters, type, topic, fromDate, toDate);

        Calendar tmpCalendar = Calendar.getInstance();
        tmpCalendar.setTime(fromDate);
        Date tmpDate = tmpCalendar.getTime();

        List<String> resultDate = new ArrayList<>();
        List<Integer> resultCount = new ArrayList<>();

        while (tmpDate.getTime() <= toDate.getTime()) {
            List<DashboardHistoryDto> tmpHistories = new ArrayList<>();
            for (DashboardHistoryDto history : dashboardHistories) {

                String d1 = sdf.format(history.getRecordDate().getTime());
                String d2 = sdf.format(tmpDate.getTime());

                if (StringUtils.equals(d1, d2)) {
                    tmpHistories.add(history);
                }
            }
            int count = 0;
            for (DashboardHistoryDto sumHistory : tmpHistories) {
                Map content = JSON.parseObject(sumHistory.getContent(), Map.class);
                Integer tmpCount = (Integer) content.get("domainCount");
                count += tmpCount;
            }
            resultDate.add(sdf.format(tmpDate));
            resultCount.add(count);

            tmpCalendar.add(Calendar.DATE, 1);
            tmpDate = tmpCalendar.getTime();
        }

        Map<String, List> result = new HashMap<>(2);
        result.put("xAxis", resultDate);
        result.put("yAxis", resultCount);

        return result;
    }

    @Override
    public Map<String, List> getExecutorHistory(String zkCluster, String type, String topic, Date fromDate,
                                                Date toDate) {

        SimpleDateFormat sdf = new SimpleDateFormat("MM-dd");

        List<String> zkClusters = new ArrayList<>();

        if (StringUtils.isEmpty(zkCluster)) {
            zkClusters.addAll(getAvailableZkClusterKey());
        } else {
            zkClusters.add(zkCluster);
        }

        if (zkClusters.isEmpty()) {
            Map<String, List> result = new HashMap<>(2);
            result.put("xAxis", null);
            result.put("yAxis", null);
            return result;
        }

        List<DashboardHistoryDto> dashboardHistories = dashboardHistoryDtoRepository
                .selectByZkClustersAndTypeAndTopicAndFromStartDateToEndDate(zkClusters, type, topic, fromDate, toDate);

        Calendar tmpCalendar = Calendar.getInstance();
        tmpCalendar.setTime(fromDate);
        Date tmpDate = tmpCalendar.getTime();

        List<String> resultDate = new ArrayList<>();
        List<Integer> otherCountList = new ArrayList<>();
        List<Integer> dockerCountList = new ArrayList<>();
        List<Integer> totalCountList = new ArrayList<>();

        while (tmpDate.getTime() <= toDate.getTime()) {
            List<DashboardHistoryDto> tmpHistories = new ArrayList<>();
            for (DashboardHistoryDto history : dashboardHistories) {

                String d1 = sdf.format(history.getRecordDate().getTime());
                String d2 = sdf.format(tmpDate.getTime());

                if (StringUtils.equals(d1, d2)) {
                    tmpHistories.add(history);
                }
            }
            int otherCount = 0;
            int dockerCount = 0;
            int totalCount = 0;
            for (DashboardHistoryDto sumHistory : tmpHistories) {
                Map content = JSON.parseObject(sumHistory.getContent(), Map.class);
                Integer other = (Integer) content.get("otherCount");
                otherCount += other;
                Integer docker = (Integer) content.get("dockerCount");
                dockerCount += docker;
                Integer total = other + docker;
                totalCount += total;
            }
            resultDate.add(sdf.format(tmpDate));


            otherCountList.add(otherCount);
            dockerCountList.add(dockerCount);
            totalCountList.add(totalCount);

            tmpCalendar.add(Calendar.DATE, 1);
            tmpDate = tmpCalendar.getTime();
        }


        Map<String, List> resultData = new HashMap<>(4);
        resultData.put("date", resultDate);
        resultData.put("dockerCount", dockerCountList);
        resultData.put("otherCount", otherCountList);
        resultData.put("totalCount", totalCountList);

        return resultData;
    }

    @Override
    public Map<String, List> getJobCountHistory(String zkCluster, String type, String topic, Date fromDate,
                                                Date toDate) {
        SimpleDateFormat sdf = new SimpleDateFormat("MM-dd");

        List<String> zkClusters = new ArrayList<>();

        if (StringUtils.isEmpty(zkCluster)) {
            zkClusters.addAll(getAvailableZkClusterKey());
        } else {
            zkClusters.add(zkCluster);
        }

        if (zkClusters.isEmpty()) {
            Map<String, List> result = new HashMap<>(2);
            result.put("xAxis", null);
            result.put("yAxis", null);
            return result;
        }

        List<DashboardHistoryDto> dashboardHistories = dashboardHistoryDtoRepository
                .selectByZkClustersAndTypeAndTopicAndFromStartDateToEndDate(zkClusters, type, topic, fromDate, toDate);

        Calendar tmpCalendar = Calendar.getInstance();
        tmpCalendar.setTime(fromDate);
        Date tmpDate = tmpCalendar.getTime();

        List<String> resultDate = new ArrayList<>();
        List<Integer> resultCount = new ArrayList<>();

        while (tmpDate.getTime() <= toDate.getTime()) {
            List<DashboardHistoryDto> tmpHistories = new ArrayList<>();
            for (DashboardHistoryDto history : dashboardHistories) {

                String d1 = sdf.format(history.getRecordDate().getTime());
                String d2 = sdf.format(tmpDate.getTime());

                if (StringUtils.equals(d1, d2)) {
                    tmpHistories.add(history);
                }
            }
            int count = 0;
            for (DashboardHistoryDto sumHistory : tmpHistories) {
                Map content = JSON.parseObject(sumHistory.getContent(), Map.class);
                Integer tmpCount = (Integer) content.get("jobCount");
                count += tmpCount;
            }
            resultDate.add(sdf.format(tmpDate));
            resultCount.add(count);

            tmpCalendar.add(Calendar.DATE, 1);
            tmpDate = tmpCalendar.getTime();
        }

        Map<String, List> result = new HashMap<>(2);
        result.put("xAxis", resultDate);
        result.put("yAxis", resultCount);

        return result;
    }

    private List<String> getAvailableZkClusterKey() {
        List<String> zkClusters = new ArrayList<>();
        Collection<ZooKeeperCluster> zkClusterList = zooKeeperService.getOnlineZooKeeperClusterList();
        Iterator<ZooKeeperCluster> iter = zkClusterList.iterator();
        while (iter.hasNext()) {
            ZooKeeperCluster tmpCluster = iter.next();
            zkClusters.add(tmpCluster.getZooKeeperClusterKey());
        }
        return zkClusters;
    }

    public enum DashboardType {
        DOMAIN, EXECUTOR, JOB
    }

    public enum DashboardTopic {
        DOMAIN_OVERALL_COUNT, DOMAIN_COUNT, EXECUTOR_COUNT, JOB_COUNT
    }
}
